package com.abyss.transformation;

import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.GroupReduceOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 读取apache.log日志，统计ip地址访问pv数量，使用 reduceGroup 操作聚合成一个最终结果
 * 结果类似：
 * (86.149.9.216,1)
 * (10.0.0.1,7)
 * (83.149.9.216,6)
 */
public class ReduceGroupDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. File source
        DataSource<String> fileSource = env.readTextFile("/Users/abyss/Dev/toys/flink/H-flink-learn/src/main/resources/apache.log");

        // 3. 转变成元组
        MapOperator<String, Tuple2<String, Integer>> tuple = fileSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] strings = value.split(" ");
                return Tuple2.of(strings[0], 1);
            }
        });

        // 4. 使用reduceGroup方法来完成聚合计算
        GroupReduceOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> result = tuple.groupBy(0).reduceGroup(new GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public void reduce(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out) throws Exception {
                int sum = 0;
                String key = null;
                for (Tuple2<String, Integer> ele : values) {
                    sum += ele.f1;
                    key = ele.f0;
                }
                out.collect(Tuple2.of(key, sum));
            }
        });

        // 5. print
        result.print();
    }
}
